package io.github.shenbinglife.jlock.hdfs;

import io.github.shenbinglife.jlock.LockUtils;
import io.github.shenbinglife.jlock.exception.LockFailedException;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor.AbortPolicy;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.util.Daemon.DaemonFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * JAsyncHadoopFileLock
 *
 * @author shenbing
 * @date 2022/4/21
 */
public class JAsyncHadoopFileLock extends JHadoopFileLock {

  private static final Logger logger = LoggerFactory.getLogger(JAsyncHadoopFileLock.class);

  private static final int coreSize = Integer.parseInt(
      System.getProperties().getProperty("jlock.scheduler.cores", "1"));

  private static final ScheduledThreadPoolExecutor pool = new ScheduledThreadPoolExecutor(
      coreSize,
      new DaemonFactory(),
      new AbortPolicy()
  );

  /**
   * 文件锁过期时间
   */
  protected long timeoutMillis = 60_000L;
  protected long refreshPeriodMillis = 5000L;
  protected ScheduledFuture refreshTask;

  public JAsyncHadoopFileLock(Path lockFile) {
    super(lockFile);
  }

  public JAsyncHadoopFileLock(Path lockFile, long timeoutMillis) {
    super(lockFile);
    this.timeoutMillis = timeoutMillis;
  }

  public JAsyncHadoopFileLock(Path lockFile, long timeoutMillis, long refreshPeriodMillis) {
    super(lockFile);
    this.timeoutMillis = timeoutMillis;
    this.refreshPeriodMillis = refreshPeriodMillis;
  }

  @Override
  protected boolean tryLockOnce() {
    if (isLockFileExists() && isExpired()) {
      try {
        fs.delete(lockFile, true);
      } catch (IOException e) {
        throw new LockFailedException("Can not delete expired lock file", e);
      }
    }

    boolean success = super.tryLockOnce();
    if (success) {
      LockRefreshTask task = new LockRefreshTask(lockFile, LockUtils.currentClientId());
      this.refreshTask = pool.scheduleWithFixedDelay(
          task,
          0L,
          refreshPeriodMillis,
          TimeUnit.MILLISECONDS);
    }
    return success;
  }

  @Override
  public void unlock() {
    this.refreshTask.cancel(true);

    if (isLocked()) {
      super.unlock();
    } else {
      logger.warn("Lock may expire when unlock, we need care it's safety");
    }
  }

  @Override
  public boolean isLocked() {
    return isLockFileExists() && !isExpired();
  }

  public boolean isExpired() {
    if (isLockFileExists()) {
      try {
        FileStatus fileStatus = fs.getFileStatus(lockFile);
        return System.currentTimeMillis() - fileStatus.getModificationTime() > timeoutMillis;
      } catch (IOException e) {
        throw new LockFailedException("IOException when check lock file state", e);
      }
    } else {
      throw new LockFailedException("Can not check expire when lock file not exists");
    }
  }

  public boolean isLockFileExists() {
    return super.isLocked();
  }

  /**
   * LockRefreshTask
   * <p>
   * 定期检查LockFile文件中的ClientId是否匹配注册时的ClientId，如果匹配，则更新LockFile的更新时间，保证该文件锁有效
   */
  public static class LockRefreshTask implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(LockRefreshTask.class);

    private Path path;
    private String clientId;
    private AtomicLong times = new AtomicLong(0L);
    private boolean stopped = false;

    public LockRefreshTask(Path path, String clientId) {
      this.path = path;
      this.clientId = clientId;
    }

    @Override
    public void run() {
      logger.debug("refresh lock: {}", path);
      try {
        FileSystem fs = path.getFileSystem(new Configuration());
        if (times.getAndIncrement() % 3 == 0) {
          try (InputStream in = fs.open(path);
              Reader inputReader = new InputStreamReader(in, StandardCharsets.UTF_8);
              BufferedReader reader = new BufferedReader(inputReader)) {
            String curId = reader.readLine();
            if (!Objects.equals(curId, this.clientId)) {
              logger.warn("Current file lock client id not equals origin client id, stop keep");
              stopped = true;
            }
          } catch (IOException e) {
            throw new LockFailedException("Can not read client id from file", e);
          }
        }

        if (!stopped) {
          fs.setTimes(path, System.currentTimeMillis(), System.currentTimeMillis());
        }
      } catch (IOException e) {
        logger.warn("Can not fresh lock file: " + path, e);
      }
    }
  }

}
